package com.study.flink.java.day04_async;

import com.study.flink.java.day04_async.async_io.http.BalanceAsyncToActivityBeanFunction;
import com.study.flink.java.day04_async.sink.mysql.MysqlSink;
import com.study.flink.java.day04_async.sink.redis.RedisActivityBeanMapper;
import com.study.flink.java.utils.FlinkUtilsDemo;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;

import java.util.concurrent.TimeUnit;

/**
 * 按照唯独信息统计,统计信息下沉到自定义mysql sink中
 // data example: u001,A1,2020-02-26 10:12:13,1,北京市
 // data example: u002,A2,2020-02-25 10:11:12,1,厦门市
 // data example: u003,A3,2020-02-25 20:44:55,1,泉州市
 // data example: u003,A2,2020-02-25 18:44:19,3,泉州市
 // data example: u003,A4,2020-02-23 08:28:32,2,莆田市
 // data example: u003,A1,2020-02-23 09:28:17,3,莆田市
 // Running Args: wc10 gid-wc10 node02:9092
 *
 */
public class ActivityCount {
    public static void main(String[] args) throws Exception {

        // 创建Env和添加Source
        DataStream<String> lines = FlinkUtilsDemo.createKafkaStream(args, new SimpleStringSchema());

        //Transformation
        SingleOutputStreamOperator<ActivityBean> beans = AsyncDataStream.unorderedWait(lines, new BalanceAsyncToActivityBeanFunction(), 5000, TimeUnit.MILLISECONDS, 10);

        //Transformation
        //SingleOutputStreamOperator<ActivityBean> summed1 = beans.keyBy("aid", "eventType").sum("count");
        SingleOutputStreamOperator<ActivityBean> summed2 = beans.keyBy("aid", "eventType", "province").sum("count");

        //Sink
        //summed1.addSink(new MysqlSink());
        // 创建一个redis conf
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("node02").setPassword("redis123456").setDatabase(3).build();
        summed2.addSink(new RedisSink<ActivityBean>(conf, new RedisActivityBeanMapper()));

        FlinkUtilsDemo.getEnv().execute("ActivityCount");
    }





}
